# Copyright (C) 2009 ApplianSys Ltd.
# 
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, 
# MA  02110-1301, USA.

"""
This is file contains the definition of the Worthyd class.  This is all the
important parts of the daemon, with the exception of the command line options
processing.

The intention is that worthyd can be embedded into another python programme,
though most uses will be via the commandline.
"""

import logging
import sys
import time
import threading
import socket
import os
import Queue

from logging.handlers import SysLogHandler
from ConfigParser import ConfigParser, NoOptionError

import ipaddr
import nodes
import udp

from command_timeout import command_timeout, TimeoutException


# various constants
TEN_MILLISECONDS = 10
MAX_MISSING = 2
DEFAULT_GROUP = "default_group"
DEFAULT_PORT = 8555  # unused as of Ubuntu 8.10 /etc/services

# backoff algorithms
ASYMPTOTIC = "asymptotic"
LOGARITHMIC = "logarithmic"
BACKOFF_ALGORYTHMS = [ASYMPTOTIC, LOGARITHMIC]

# worthiness functions
LOWEST_ADDRESS = "lowest_address"
HIGHEST_ADDRESS = "highest_address"
WORTHINESS_FUNCTIONS = [LOWEST_ADDRESS, HIGHEST_ADDRESS]

# other constants
FOUR_BILLION = 2**32
MAX_NAME_LEN = 300
MAX_UDP_LEN = 2**12     # maximum size of udp packet to accept
CONFIG_DIR = '/etc/worthyd'
CONFIG_FILENAME = 'worthyd.conf'
DEFAULT_TIMEOUT = 3     # give commands three seconds to execute, by default



def string_to_ipaddr(address_str):
    """
    This converts a string to *either* an ipv4 or ipv6 address.
    """
    address = None
    try:
        address = ipaddr.IPv4(address_str)
    except ipaddr.IPv4IpValidationError, e:
        pass
    except ValueError, e:
        pass
    try:
        address = ipaddr.IPv6(address_str)
    except ipaddr.IPv6IpValidationError, e:
        pass
    except ValueError, e:
        pass
    if address == None:
        return None
    return address


class WorthydException(Exception):
    """
    This class just subclasses Exception for the purpose of making exceptions
    which can be handled cleanly.
    """

    pass


class Worthyd(object):
    """
    This class is the Worthy Daemon.
    Multiple worthyds can run run on one machine, but only one should run on
    and particular address.
    """
    def __init__(self, 
                 group_names, 
                 interval, 
                 max_missing, 
                 backoff_algorithm,
                 worthiness_function, 
                 port,
                 address,
                 config_dir,
                 logger,
                 ):
        self.group_names = group_names
        self.interval = interval 
        self.max_missing = max_missing 
        self.backoff_algorithm = backoff_algorithm
        self.worthiness_function = worthiness_function 
        self.port = port
        self.address = address
        self.config_dir = config_dir
        self.logger = logger
        self.runtime = 0    # runtime is measured in ms

        if type(address) == ipaddr.IPv6:
            sys.stderr.write("ERROR: IPv6 support not yet implemented, though it has been in mind while designing the protocol / udp packet structure.\n")
            sys.exit(1)

        # these three variables are monitored by their respective threads
        self.broadcasting = None
        self.listening = None
        self.executing = None
    
        # intialise the store
        groups = {}
        for g in self.group_names:
            groups[g] = nodes.Group(g)
        self.store = nodes.Store(groups)

        # create missing directory structure
        self._create_dir_structure()

        # read in the config
        self._read_config()

        # create a queue of (command, enviroment) tuples to execute
        self.death_row = Queue.Queue()
        

    def _create_dir_structure(self):
        """
        This creates all needed directory structure, if any are missing.
        """

        try:
            os.stat(self.config_dir)
        except OSError:
            # the dir does not exist, so try to create it or fail
            try:
                os.mkdir(self.config_dir)
            except OSError:
                # cannot create, panic
                raise WorthydException("Unable to create %s" % (self.config_dir,))
            

    
    def _read_config(self):
        """
        This method reads the config file.
        """
        
        try:
            self.config = ConfigParser()
            filename = os.path.join(self.config_dir, CONFIG_FILENAME)
            self.config.read(filename)
        except Exception, e:
            raise WorthydException("Unable to read %s" % (filename,))
        
        self.logger.info(self.config)
        

    def cleanup(self):
        """
        This removes all the files when the daemon exits.
        """

        for group_name, group in self.store.groups.items():
            filename = os.path.join(self.config_dir, group_name + "-members")
            filename1 = os.path.join(self.config_dir, group_name + "-elected")
            self.logger.info("%s %s" % (filename, filename1))
            try:
                os.remove(filename)
                os.remove(filename1)
            except Exception, e:
                pass    # don't let failure to delete one file
                        # stop the deletion of the rest
    

    def highest_address(self):
        """
        This method returns a worthiness based on the address.
        """
        
        return int(self.address)


    def lowest_address(self):
        """
        Returns the opposite of lowest_address.
        """

        return FOUR_BILLION - self.highest_address()


    def asymptotic_backoff(self, group):
        """
        This returns the number of intervals to wait between sending 
        broadcasts.
        """

        return ((2 ** group.rank(self.address,
                                 self.runtime,
                                 )) 
                * self.interval
                )
        

    def logarithmic_backoff(self, group):
        """
        This returns the number of intervals to wait between sending 
        broadcasts.
        """

        return ((1 + group.rank(self.address,
                                self.runtime,
                                )) 
                * self.interval
                )


    def check_for_membership_change(self, group):
        """
        This checks to see if the membership of a group has changed.
        """

        if group.md5(self.runtime) == group.last_md5:
            return    # no change
        self.on_membership_change(group)
        group.last_md5 = group.md5(self.runtime)


    def on_membership_change(self, group):
        """
        This is called if the group's membership has changed.
        """
        
        self.logger.info("MEMBERSHIP: in group '%s'"
                         % (group.group_name,))
        addresses = " ".join([str(x) for x in group.valid_addresses(self.runtime)])
        addresses_nl = "\n".join([str(x) for x in group.valid_addresses(self.runtime)])
        self.logger.info("MEMBERSHIP %s: %s" 
                         % (group.group_name,
                            addresses,
                            )
                         )
        try:
            command = self.config.get(group.group_name, 'membership')
            env = dict(os.environ)

            # tell the command about the group members
            if addresses:
                env['WORTHYD_MEMBERS'] = "%s" % (addresses,)

            self.logger.debug("command added to death row: '%s'"
                              % (command))
            self.death_row.put((command, env))
        except NoOptionError, e:
            # do nothing if the option doesn't exist
            self.logger.info("no command configured for execution on membership change")

        # change the text file
        filename = "%s-members" % (group.group_name,)
        try:
            file = open(os.path.join(self.config_dir, filename), "w")
            file.write(addresses_nl)
            file.close()
        except Exception, e:
            self.logger.warn("error writing to %s" % (filename,))


    def check_for_most_worthy_change(self, group):
        """
        This checks to see if the most worthy node has changed.
        """
        #self.logger.debug("checking for changes to most_worthy")

        most_worthy = group.most_worthy(self.runtime)
        if most_worthy and group.last_most_worthy:
            if group.last_most_worthy.address == most_worthy.address:
                return
        else:
            if (group.last_most_worthy == None 
                and most_worthy == None):
                return

        # otherwise there has been a change
        if (most_worthy 
            and most_worthy.address == self.address):
            self.on_most_worthy(group)
        if (group.last_most_worthy 
            and group.last_most_worthy.address
            == self.address):
            self.on_unworthy(group)
        self.on_most_worthy_change(group)
        group.last_most_worthy = most_worthy


    def on_most_worthy_change(self, group):
        """
        This method is called whenever the most worthy node changes.
        """

        most_worthy = group.most_worthy(self.runtime)
        if most_worthy:
            self.logger.info("%s is now most worthy in '%s'"
                             % (most_worthy.address, group.group_name))
        else:
            self.logger.warn("there is now no most worthy node in '%s'"
                             % (group.group_name,))

        # change the text file
        filename = "%s-elected" % (group.group_name,)
        try:
            file = open(os.path.join(self.config_dir, filename), "w")
            if most_worthy:
                file.write("%s" % (most_worthy.address,))
            else:
                file.write("")
            file.close()
        except Exception, e:
            self.logger.warn("error writing to %s" % (filename,))



    def on_most_worthy(self, group):
        """
        This method is called when this node becomes 'most worthy' for a group.
        """

        self.logger.info("ELECTED: this node (%s) has become most worthy for '%s'"
                         % (self.address, group.group_name))
        try:
            command = self.config.get(group.group_name, 'elected')
            env = dict(os.environ)
            self.logger.debug("command added to death row: '%s'"
                             % (command,))
            self.death_row.put((command, env))
        except NoOptionError, e:
            # do nothing if the option doesn't exist
            self.logger.info("no command configured for execution on election")


    def on_unworthy(self, group):
        """
        This method is called when this node stops being 'most worthy' for a 
        group.
        """

        self.logger.info("DEPOSED: this node (%s) is no longer most worthy for '%s'"
                         % (self.address, group.group_name))
        try:
            command = self.config.get(group.group_name, 'deposed')
            env = dict(os.environ)
            most_worthy = group.most_worthy(self.runtime)

            # if there is an elected node, then put it in the environment
            if most_worthy:
                env['WORTHYD_ELECTED'] = "%s" % (most_worthy.address,)

            self.logger.debug("command added to death row: '%s'"
                             % (command))
            self.death_row.put((command, env))
        except NoOptionError, e:
            # do nothing if the option doesn't exist
            self.logger.info("no command configured for execution on being deposed")


    def start(self):
        """
        This method is called to start the deamon running.
        """

        if True: # it's useful to have buggy code running in the original thread
            broadcast_thread = threading.Thread(target=self.broadcast_main)
            executioner_thread = threading.Thread(target=self.executioner_main)
            try:
                broadcast_thread.start()
                executioner_thread.start()   # do the executing in this thread
                self.listener_main()   # do the listening in this thread
            except Exception, e:
                raise e
            finally:
                self.broadcasting = False
                self.executing = False
                broadcast_thread.join()  # wait for the listener to finish
                executioner_thread.join()  # wait for the executioner to finish
        else:
            pass
            """
            listener_thread = threading.Thread(target=self.listener_main)
            try:
                listener_thread.start()
                self.broadcast_main()   # do the broadcasting in this thread
            except Exception, e:
                raise e
            finally:
                self.listening = False
                listener_thread.join()  # wait for the listener to finish
            """

    def broadcast_main(self):
        """
        This method is the 'main' method of the broadcast thread.
        """
        
        self.broadcasting = True
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) 

        # add this machine's node to the store
        for group_name, group in self.store.groups.items():
            node = nodes.Node(self.address, 
                              0,        # start as unworthy
                              None,     # a node cannot expire from itself
                              "x" * 16, # no md5sum yet
                              )
            group.add_node(node)

        while self.broadcasting:
            time.sleep(0.001 * self.interval)
            #self.logger.debug("broadcasting")

            # manage time
            self.runtime += self.interval
            self.logger.debug("")
            self.logger.debug("runtime >= %sms" % (self.runtime,))

            self.store.lock.acquire()
            try:
                for group_name, group in self.store.groups.items():
                    #self.logger.debug("--------debug---------")
                    #self.logger.debug("broadcasting for %s" % (group_name,))
                    #self.logger.debug("current store %s" % (self.store,))
                    #self.logger.debug("runtime %s" % (self.runtime,))
                    #self.logger.debug("valid_nodes %s" % (group.valid_nodes(self.runtime),))
                    #self.logger.debug("md5 %s" % (group.md5(self.runtime).__repr__()))
                    #self.logger.debug("most_worthy: %s" % group.valid_nodes(self.runtime)[0].address)
                    #self.logger.debug("-----end-of-debug-----")

                    # check for changes (due to self.runtime being incremented)
                    self.check_for_most_worthy_change(group)
                    self.check_for_membership_change(group)

                    worthiness = None
                    if self.worthiness_function == HIGHEST_ADDRESS:
                        worthiness = self.highest_address()
                    if self.worthiness_function == LOWEST_ADDRESS:
                        worthiness = self.lowest_address()
                    if group.starting:
                        worthiness = 0  # surpress worthiness
                    assert worthiness != None, "logical error in worthiness"

                    broadcast_interval = None
                    if self.backoff_algorithm == LOGARITHMIC:
                        broadcast_interval = self.logarithmic_backoff(group)
                    if self.backoff_algorithm == ASYMPTOTIC:
                        broadcast_interval = self.asymptotic_backoff(group)
                    assert broadcast_interval != None, "logical error in interval"

                    valid_for = broadcast_interval * self.max_missing

                    # add the node to it's own group
                    group.add_node(nodes.Node(self.address, 
                                         worthiness,
                                         None,  # a node cannot expire from itself
                                         group.md5(self.runtime), # no md5sum yet
                                         ))

                    # turn off initialisation if the group is stable
                    if group.stable(self.runtime):
                        group.starting = False

                    # if the group is not starting, then only broadcast when we must
                    if not group.starting:
                        if not group.time_to_broadcast(broadcast_interval, self.runtime):
                            #self.logger.debug("not time to broadcast yet")
                            continue
                        else:
                            self.logger.debug("broadcasting '%s'" 
                                              % (group.group_name,))
                            group.last_broadcast = self.runtime

                    packet = udp.BroadcastPacket(worthiness=worthiness,
                                                 num_members=group.count(self.runtime),
                                                 members_md5=group.md5(self.runtime),
                                                 valid_for=valid_for,
                                                 group_name=group_name,
                                                 )

                    #self.logger.debug("packet %s, %s", 
                    #                  packet.buffer.__repr__(), 
                    #                  "255.255.255.255",
                    #                  )
                    sock.sendto(packet.buffer, ("255.255.255.255", self.port))
            finally:
                self.store.lock.release()

        self.logger.debug("broadcaster finished")
    

    def listener_main(self):
        """
        This method is the 'main' method of the listener thread.
        """

        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        #sock.bind((str(self.address).split("/")[0], self.port))
        sock.bind(('', self.port)) # TODO: make this listen on a particular broadcast
        self.listening = True
        while self.listening:
            #self.logger.debug("listening")
            data, (addr, port) = sock.recvfrom(MAX_UDP_LEN)

            # ignore packets broadcast from this node
            if ipaddr.IPv4(addr) == self.address:
                continue

            #self.logger.debug("received %s from %s" % (data.__repr__(), addr))
            try:
                packet = udp.BroadcastPacket(buffer=data, sender=addr)
            except udp.PacketException, e:
                self.logger.warn("Invalid packet received from %s." % (addr,))
                self.logger.debug("%s" % e)

            # is this a packet for a group in which we have no interest?
            if not self.store.groups.has_key(packet.group_name):
                self.logger.debug("ignoring packet for group %s from %s" % (packet.group_name, addr))
                continue

            # otherwise we need to add send this packet to the store
            self.logger.debug("recieved packet for group %s from %s" % (packet.group_name, addr))
            self.store.lock.acquire()
            try:
                self.store.receive_packet(packet, self.runtime)
                self.check_for_most_worthy_change(self.store.groups[packet.group_name])
                self.check_for_membership_change(self.store.groups[packet.group_name])
            finally:
                self.store.lock.release()

        self.logger.debug("listener finished")


    def executioner_main(self):
        """
        This method is the 'main' method of the executioner thread.
        It simply reads (command, environment) tuples from the queue and 
        executes them.
        """
    
        self.logger.debug("executioner started")

        self.executing = True
        while self.executing:
            # Wait for one second for a command to come in.
            # This adds a one second delay before the programme will exit 
            # on ctrl-C.
            try:
                command, env = self.death_row.get(True, 1)
                self.logger.info("Executing '%s''." % (command,))
                try:
                    exit, out, err = command_timeout(["bash", "-c", "%s" 
                                                      % command], 
                                                     timeout=DEFAULT_TIMEOUT,
                                                     env=env,
                                                     )
                    self.logger.info("Executed '%s'." % (command,))
                    if exit == 0:
                        self.logger.info("exit code: %s, stdout: '%s' "
                                         "stderr: '%s'." 
                                         % (exit, out, err))
                    else:
                        self.logger.warn("exit code: %s, stdout: '%s' "
                                         "stderr: '%s'." 
                                         % (exit, out, err))
                except TimeoutException, e:
                    self.logger.warn("Execution of '%s' timed out after %ss." 
                                     % (command, DEFAULT_TIMEOUT))
            except Queue.Empty, e:
                pass    # just loop

        if not self.death_row.empty():
            self.logger.warning("%s commands left in the queue unexecuted." 
                                % (self.death_row.qsize())
                                )

        self.logger.debug("executioner finished")

